package cn.com.utt.common.protorpc.codec;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.codec.binary.Hex;

import com.google.common.base.Preconditions;
import com.google.protobuf.Message;

import cn.com.utt.common.protorpc.message.Callback;
import cn.com.utt.common.protorpc.message.RPCMessage;
import cn.com.utt.common.protorpc.proto.ReplyProto;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.CorruptedFrameException;

public class MessageDecoder extends ByteToMessageDecoder {

	private final AtomicLong seqNum = new AtomicLong(1L);

	private final Map<Long, Callback> callbackMap = new ConcurrentHashMap<Long, Callback>();

	public long getNextSeqId() {
		return seqNum.getAndIncrement();
	}

	public synchronized void registerCallback(long seqId, Callback callback) {
		if (callbackMap.containsKey(seqId)) {
			throw new IllegalArgumentException("Callback already registered");
		}
		callbackMap.put(seqId, callback);
	}

	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		in.markReaderIndex(); // 标记开始读的位置
		int preIndex = in.readerIndex();
		System.out.println("decode");
		// decode response
		RPCMessage message = new RPCMessage();
		// read header
		ByteBuf buf = in.copy();
		int n = buf.readableBytes();
		byte[] array = new byte[n];
		buf.readBytes(array);
		System.out.println("resp readbuf:" + Hex.encodeHexString(array));
		// header len
		long hlen = readRawVarint64(in);
		if (preIndex == in.readerIndex()) {
			// 没有读取长度数据，或读取出错
			return;
		}
		if (hlen < 0) {
			throw new CorruptedFrameException("negative length: " + hlen);
		}
		System.out.println("read:" + hlen);
		if (in.readableBytes() < hlen) {
			// 数据长度不够，读取index复位，等待更多数据
			System.out.println("reset");
			in.resetReaderIndex();
			return;
		}
		byte[] headerArray = new byte[(int) hlen];
		in.readBytes(headerArray);
		// proto decode
		ReplyProto.Reply response = ReplyProto.Reply.parseFrom(headerArray);
		System.out.println(response.toString());
		message.setHeader(response);
		// read body

		preIndex = in.readerIndex();
		long blen = readRawVarint64(in);
		System.out.println("body len:" + blen);
		if (preIndex == in.readerIndex()) {
			// 没有读取长度数据，或读取出错
			in.resetReaderIndex();
			return;
		}
		if (blen < 0) {
			throw new CorruptedFrameException("negative length: " + blen);
		}
		if (in.readableBytes() < blen) {
			// 数据长度不够，读取index复位，等待更多数据
			System.out.println("reset");
			in.resetReaderIndex();
			return;
		}
		System.out.println("resp seqId:" + (response.getSeq() == 0));
		long seqId = response.getSeq();
		System.out.println("resp seqId:" + seqId);
		byte[] bodyArray = new byte[(int) blen];
		in.readBytes(bodyArray);

		Callback callback = callbackMap.remove(seqId);
		if (callback == null) {
			System.out.println("Received response with no callback registered");
		} else {
			if (response.getError().length() > 0) {
				// throw error
				callback.setError(response.getError());
			}
			if (callback.getReply() != null) {
				// System.out.println("Invoking callback with response");
				// proto decode
				Message reply = callback.getReply().getParserForType().parseFrom(bodyArray);
				System.out.println("call back reply:" + reply.toString());
				message.setBody(reply);
				// System.out.println("after setBody");
				callback.run(reply);
			} else {
				// System.out.println("reply null");
				// System.out.println(callback);
				callback.run(null);
			}
		}
		out.add(message);
	}

	public static long readRawVarint64(ByteBuf buffer) throws IOException {
		if (!buffer.isReadable()) {
			return 0;
		}
		long raw = readUnsignedVarLong(buffer);
		// This undoes the trick in writeSignedVarLong()
		long temp = (((raw << 63) >> 63) ^ raw) >> 1;
		// This extra step lets us deal with the largest signed values by
		// treating
		// negative results from read unsigned methods as like unsigned values
		// Must re-flip the top bit if the original read value had it set.
		return temp ^ (raw & (1L << 63));
	}

	public static long readUnsignedVarLong(ByteBuf buffer) throws IOException {
		long value = 0L;
		int i = 0;
		long b;
		if (!buffer.isReadable()) {
			buffer.resetReaderIndex();
			return 0;
		}
		b = buffer.readByte();
		while ((b & 0x80L) != 0) {
			value |= (b & 0x7F) << i;
			i += 7;
			Preconditions.checkArgument(i <= 63, "Variable length quantity is too long (must be <= 63)");
			if (!buffer.isReadable()) {
				buffer.resetReaderIndex();
				return 0;
			}
			b = buffer.readByte();
		}
		return value | (b << i);
	}

}